package edu.upenn.seas.peerreview.transport

import scala.actors._
import scala.actors.Actor._
import scala.actors.remote._
import scala.collection.mutable
import scala.util.logging.Logged

import edu.upenn.seas.peerreview.{Id, Source}

/**
 * A remote (multi-JVM, centralized) transport layer's client component. The
 * path of a message is from the sink to this client, to the server, to the
 * client of the destination source, and finally to the source itself.
 */
class RemoteClient(serverIp: String, port: Int) extends Transport with Logged {

  /* Retrieving messages from the server uses client polling instead of server
   * pushing. This avoids false assumptions about RemoteActor internals (e.g.
   * what is a sender?) and networking (e.g. what about dynamic client IPs?).
   */

  import RemoteServer._
  import RemoteClient._
  import Transport._
  
  
  /* <p>This hackery is needed to use a non-default class loader when
   * deserializing objects with the server proxy. It would be nice to write
   * this code as
   * <pre>val server = RemoteActor.select(Node(serverIp, port), 'server)</pre>
   * but the class loader needs to be set correctly.
   * <p>As seen in RemoteActor source, to set class loader requires first
   * setting RemoteActor.classLoader and then calling RemoteActor.select --
   * which refers to the static variable. The loader-self pairing is saved
   * forever. Therefore we create a fresh self (a new actor) to ensure that the
   * static loader we set is the first pairing the RemoteActor sees.
   */
  // TODO: factor out "set up remote actor" code & revise the above comment
  val server: Actor = actor {
    RemoteActor.classLoader = Thread.currentThread.getContextClassLoader
    val delegate = RemoteActor.select(Node(serverIp, port), 'server)
    loop {
      react {
        case m => {
          if (!m.isInstanceOf[RequestMessages]) {
            log("sending to server " + m)
          }
          delegate.send(m, sender)
        }
      }
    }
  }
  
  /**
   * Registers sources and dispatches messages locally.
   */
  val localTransport = new LocalTransport() {
    override def log(msg: String) = RemoteClient.this.log(msg)
  }
  
  override def register(name: String, factory: Source.Factory): Id = {
    if (name == coordinatorName) {
      log("seeking ownership of the coordinator before registering coord")
      assert(server !? (timeout, RegisterCoordinator(clientId)) match {
        case Some(RegisteredCoordinator(permissionGranted)) => permissionGranted
        case m => error("Unexpected during coordinator registration: " + m
                        + ". Is the server still running?")
      }, "there's already registered coordinator in the distributed transport")
    }
    val localId = localTransport.register(name, (localId, localSink) => {
      /**
       * Don't optimize for locally-destined messages. Doing so requires testing
       * with multiple RemoteClient instances.
       */
      val remoteId = getRemoteId(clientId, localId)
      factory(remoteId, server ! SendMessage(remoteId, _, _))
    })
    val id = getRemoteId(clientId, localId)
    log("just registered " + id)
    id
  }
  
  private def enqueueMesg(from: Id, to: Id, mesg: Message) {
    val localId: Id = to match {
      case RemoteId(nid, lid) => lid
      case Id.coordinatorId => Id.coordinatorId
    }
    localTransport.enqueueMesg(from, localId, mesg)
  }
  
  // register with the server
  log("registering client with server")
  private val clientId = (server !? (timeout, RegisterClient())) match {
    case Some(RegisteredClient(cid)) => {
      log("received assignment to client id " + cid)
      cid
    }
    case m => error("Unexpected during transport client registration: " + m
                    + ". Are you sure the server is running?")
  }
  
  // poll the server for new messages
  actor {
    loop {
      (server !? (timeout, RequestMessages(clientId))) match {
        case Some(reply) => {
          reply match {
            case req: RequestedMessages => {
              req.mesgs foreach (m => enqueueMesg(m.from, m.to, m.mesg))
            }
            case InvalidClientId(badId) => {
              log("oops, the server doesn't know this client " + badId)
              exit
            }
          }
        }
        /* If connection times out, cause an error. Some messages may have been
         * dropped so just quit (until we prevent dropped messages).
         */
        case None => error("failed to connect to server...quitting")
      }
    }
  }
}

object RemoteClient {

  import RemoteServer._

  /**
   * For testing and internal use only. The timeout when making synchronous
   * calls. Timing out prevents inexplicable infinite loops.
   */
  private[transport] val timeout = 5 * 60 * 1000

  private def getRemoteId(clientId: ClientId, localId: Id): Id =
    if (localId == Id.coordinatorId) localId else RemoteId(clientId, localId)
}
